package com.gitee.melin.bee.core.jdbc.dialect.maxcompute;

import com.gitee.melin.bee.core.jdbc.dialect.DataTypeConvertor;
import com.gitee.melin.bee.core.jdbc.dialect.DefaultTypeConvertor;
import com.gitee.melin.bee.core.jdbc.dialect.AbstractJdbcDialect;
import com.gitee.melin.bee.core.jdbc.enums.DataSourceType;
import com.gitee.melin.bee.core.jdbc.dialect.IDBQuery;
import com.gitee.melin.bee.core.jdbc.relational.MetaColumn;
import com.gitee.melin.bee.core.jdbc.relational.ConnectionInfo;
import com.gitee.melin.bee.core.jdbc.relational.MetaTable;
import com.gitee.melin.bee.util.JsonUtils;
import com.google.common.collect.Lists;
import org.apache.commons.lang3.StringUtils;

import java.sql.*;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

public class MaxComputeJdbcDialect extends AbstractJdbcDialect {

    public MaxComputeJdbcDialect(ConnectionInfo connectionInfo) {
        super(connectionInfo);
    }

    public MaxComputeJdbcDialect(Connection connection) {
        super(connection);
    }

    @Override
    protected IDBQuery getDBQuery() {
        return new MaxComputeQuery();
    }

    @Override
    protected DataTypeConvertor getTypeConvert() {
        return new DefaultTypeConvertor();
    }

    @Override
    protected Driver getDriver() {
        return loadDriver(DataSourceType.MAXCOMPUTE.getDriverClass());
    }

    @Override
    public List<MetaTable> getSchemaTables(String schemaName) {
        String sql = getDBQuery().tablesSql(schemaName);
        try (Connection connection = connectionFactory.openConnection();
             PreparedStatement preparedStatement = connection.prepareStatement(sql);
             ResultSet resultSet = preparedStatement.executeQuery()) {

            List<MetaTable> tableList = new ArrayList<>();
            while (resultSet.next()) {
                String json = resultSet.getString(getDBQuery().tableName());
                String[] split = json.split("\n");
                for (String table : split) {
                    if (table.contains(":")){
                        table = table.split(":")[1].trim();
                    }

                    MetaTable tableInfo = new MetaTable();
                    tableInfo.setTableName(table);
                    tableList.add(tableInfo);
                }
            }
            return tableList;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    @Override
    public DataSourceType getDataSourceType() {
        return DataSourceType.MAXCOMPUTE;
    }

    @Override
    public MetaTable getSchemaTable(String schemaName, String tableName) {
        if (StringUtils.isBlank(schemaName)) {
            throw new IllegalStateException("schemaName can not blank");
        }

        if (StringUtils.isBlank(tableName)) {
            throw new IllegalStateException("tableName can not blank");
        }
        return this.getTable(schemaName, tableName);
    }

    protected MetaTable getTable(String schemaName, String tableName) {
        try (Connection connection = connectionFactory.openConnection()) {
            // 查询单个表信息
            String descTableSql = getDBQuery().columnsSql(schemaName, tableName);
            try (PreparedStatement statement = connection.prepareStatement(descTableSql);
                 ResultSet rs = statement.executeQuery()) {
                ResultSetMetaData metaData = rs.getMetaData();
                List<String> columnList = new ArrayList<>();
                for (int i = 1; i <= metaData.getColumnCount(); i++) {
                    columnList.add(metaData.getColumnLabel(i));
                }
                System.out.println(columnList);
                while (rs.next()) {
                    String json = rs.getString(getDBQuery().columnName());
                    Map<String, Object> map = JsonUtils.toJavaMap(json);
                    Integer createTime = (Integer) map.get("createTime");
                    String tableType = (String) map.get("tableType");

                    MetaTable tableInfo = new MetaTable();
                    tableInfo.setTableName(tableName);
                    tableInfo.setTableType(tableType);
                    tableInfo.setCreateTime(new Date(createTime * 1000L));

                    //获取列
                    Map<String, Object> sd = (Map<String, Object>) map.get("sd");
                    tableInfo.setLocation((String) sd.get("location"));
                    List<MetaColumn> columns = new ArrayList<>();
                    List<Map<String, Object>> cols = (List<Map<String, Object>>) sd.get("cols");
                    cols.forEach(c -> {
                        MetaColumn field = new MetaColumn();
                        field.setColumnName((String) c.get("name"));
                        field.setColumnType((String) c.get("type"));
                        field.setComment((String) c.get("comment"));
                        columns.add(field);
                    });
                    tableInfo.setColumns(columns);

                    //获取分区信息
                    List<Map<String, Object>> partitionKeys = (List<Map<String, Object>>)  map.get("partitionKeys");
                    List<String> partitions = Lists.newArrayList();
                    partitionKeys.forEach(p -> {
                        partitions.add((String) p.get("name"));
                    });
                    tableInfo.setPartitionKeys(partitions);
                    return tableInfo;
                }
            }
            return null;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    @Override
    public List<MetaColumn> getSchemaColumns(String schemaName, String tableName){
        try (Connection connection = connectionFactory.openConnection()) {
            // 查询单个表信息
            String descTableSql = getDBQuery().columnsSql(schemaName, tableName);
            try (PreparedStatement statement = connection.prepareStatement(descTableSql);
                 ResultSet rs = statement.executeQuery()) {
                ResultSetMetaData metaData = rs.getMetaData();
                List<String> columnList = new ArrayList<>();
                for (int i = 1; i <= metaData.getColumnCount(); i++) {
                    columnList.add(metaData.getColumnLabel(i));
                }
                System.out.println(columnList);
                while (rs.next()) {
                    String json = rs.getString(getDBQuery().columnName());
                    Map<String, Object> map = JsonUtils.toJavaMap(json);
                    //获取列
                    Map<String, Object> sd = (Map<String, Object>) map.get("sd");
                    List<MetaColumn> columns = new ArrayList<>();
                    List<Map<String, Object>> cols = (List<Map<String, Object>>) sd.get("cols");
                    cols.forEach(c -> {
                        MetaColumn field = new MetaColumn();
                        field.setColumnName((String) c.get("name"));
                        field.setColumnType((String) c.get("type"));
                        field.setComment((String) c.get("comment"));
                        columns.add(field);
                    });
                    return columns;
                }
            }
            return null;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    @Override
    protected String getDDLColumnName() {
        return "";
    }

    @Override
    public String getCreateTableScript(String schemaName, String tableName) {
        return this.getCreateHiveTableScript(schemaName, tableName);
    }

    public String getCreateHiveTableScript(String databaseName, String tableName) {
        String sql = "SHOW CREATE table " + databaseName + "." + tableName;
        try (Connection connection = connectionFactory.openConnection();
             PreparedStatement dbStat = connection.prepareStatement(sql);
             ResultSet dbResult = dbStat.executeQuery()) {

            StringBuilder createTableScript = new StringBuilder();
            while (dbResult.next()) {
                createTableScript.append(dbResult.getString(1)).append("\n");
            }
            if (createTableScript.length() > 0){
                return createTableScript.toString();
            } else {
                return "DDL is not available";
            }
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error", e);
        }
    }
}
